package com.hw

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions.{expr, map}
import org.apache.spark.sql.{Column, DataFrame, SparkSession}

object Demo7Burk {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession
      .builder()
      .config("spark.sql.shuffle.partitions", 1)
      .appName("Burk")
      .master("local")
      .getOrCreate()

    val burkDF: DataFrame = spark.read
      .format("csv")
      .option("set", ",")
      .schema("burk string,year string ,tsl01 double,tsl02 double,tsl03 double,tsl04 double,tsl05 double,tsl06 double,tsl07 double,tsl08 double,tsl09 double,tsl10 double,tsl11 double,tsl12 double")
      .load("data/burk.txt")
    //    val burkRDD: RDD[String] = spark.sparkContext.textFile("data/burk.txt")
    //    burkRDD.map(
    //      line =>{
    //        val split: Array[String] = line.split(",")
    //        ((split(0),split(1)),(split(2),split(3),split(4),split(5),split(6),split(7),split(8),split(9),split(10),split(11)))
    //      }
    //    ).flatMap(line =>{
    //      val split: Array[String] = line._2.toString().split(",")
    //      split
    //    }).foreach(println)


    import org.apache.spark.sql.functions._
    import spark.implicits._
    val tmp: Column = map(expr("1"), $"tsl01"
      , expr("2"), $"tsl02"
      , expr("3"), $"tsl03"
      , expr("4"), $"tsl04"
      , expr("5"), $"tsl05"
      , expr("6"), $"tsl06"
      , expr("7"), $"tsl07"
      , expr("8"), $"tsl08"
      , expr("9"), $"tsl09"
      , expr("10"), $"tsl10"
      , expr("11"), $"tsl11"
      , expr("12"), $"tsl12"
    )
    //1、统计每个公司每年按月累计收入
    burkDF.select($"burk", $"year", explode(tmp))
      .withColumn("month_value",sum($"value") over Window.partitionBy($"burk",$"year").orderBy("key"))
      .show()

    //2、统计每个公司当月比上年同期增长率
    burkDF.select($"burk", $"year", explode(tmp))
      .withColumn("last",lag($"value",1) over Window.partitionBy($"burk",$"key").orderBy($"year"))
      .withColumn("growth rate",round($"value"/ $"last" -1,4))
      .show()
  }

}
